package com.bw.yk02.app;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.bw.yk02.util.MyKafkaUtil;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.cep.*;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.OutputTag;

import java.util.List;
import java.util.Map;

public class UserJumpApp {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment  env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource<String> ods_page_source=env.addSource(MyKafkaUtil.getKafkaConsumer("dwd_page_log","user_jump_group"));

        SingleOutputStreamOperator<JSONObject> ods_log_filter=ods_page_source.filter(new FilterFunction<String>() {
            @Override
            public boolean filter(String s) throws Exception {
                try {
                    JSONObject jsonObject = JSON.parseObject(s);
                    if(jsonObject!=null){
                        return true;
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }
                return false;
            }
        }).map(JSON::parseObject);

        KeyedStream<JSONObject, String> keyedStream = ods_log_filter.keyBy(data -> data.getJSONObject("common").getString("mid"));

        Pattern<JSONObject,JSONObject> pattern= Pattern.<JSONObject>begin("start")
                .where(new SimpleCondition<JSONObject>() {
            @Override
            public boolean filter(JSONObject jsonObject) throws Exception {
                JSONObject jsonObject1 = jsonObject.getJSONObject("page");
                if(jsonObject1!=null&&jsonObject1.getString("last_page_id")==null){
                    return true;
                }
                 return false;
            }
        }).next("end").where(new SimpleCondition<JSONObject>() {
            @Override
            public boolean filter(JSONObject jsonObject) throws Exception {
                JSONObject jsonObject1 = jsonObject.getJSONObject("page");
                if(jsonObject1!=null&&jsonObject1.getString("last_page_id")==null){
                    return true;
                }
                return false;
            }
        }).within(Time.seconds(10L));;


        PatternStream<JSONObject> patternStream= CEP.pattern(keyedStream,pattern);

        OutputTag<JSONObject> timedOutPartialMatchesTag=new OutputTag<JSONObject>("timedOutPartialMatches"){};

        SingleOutputStreamOperator<JSONObject> result_ds=patternStream.select(timedOutPartialMatchesTag,
                new PatternTimeoutFunction<JSONObject,JSONObject>()
                {
                    public JSONObject timeout(Map<String, List<JSONObject>> pattern, long timeoutTimestamp) throws Exception {
                        List<JSONObject> start=pattern.get("0");
                        return start.get(0);
                    }
                },
                new PatternSelectFunction<JSONObject,JSONObject>() {
                    @Override
                    public JSONObject select(Map<String, List<JSONObject>> map) throws Exception {
                        List<JSONObject> start=map.get("0");
                        return start.get(0);
                    }
                }
        );


        DataStream<JSONObject> timeout_result_ds1=result_ds.getSideOutput(timedOutPartialMatchesTag);

        DataStream<JSONObject> result=result_ds.union(timeout_result_ds1);

        result.map(new MapFunction<JSONObject, String>() {
            @Override
            public String map(JSONObject jsonObject) throws Exception {
                return jsonObject.toJSONString();
            }
        }).addSink(MyKafkaUtil.getKafkaProducer("dwm_user_jump"));

        //

        env.execute("User Jump App");

    }
}
